package cn.doitedu.rtdw.data_sync;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/7
 * @Desc: 学大数据，到多易教育
 * 视频信息表同步到hbase任务
 **/
public class SyncJob02_VideoInfoTable2Hbase {

    public static void main(String[] args) {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        // 状态后端
        env.setStateBackend(new HashMapStateBackend());

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建cdc连接器表，映射MySQL中视频信息表： cms_video
        tenv.executeSql(
                "CREATE TABLE video_mysql (    " +
                        "      id BIGINT," +
                        "      video_name STRING,                 " +
                        "      video_type STRING,                 " +
                        "      video_album STRING,                " +
                        "      video_author STRING,               " +
                        "      video_timelong BIGINT,             " +
                        "      create_time timestamp(3),          " +
                        "      update_time timestamp(3),          " +
                        "     PRIMARY KEY (id) NOT ENFORCED            " +
                        "     ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw',          " +
                        "     'table-name' = 'cms_video'              " +
                        ")"
        );

        // 创建hbase连接器表，映射 hbase中的视频信息维表：dim_video_info
        //  hbase中的 rowkey是视频id补齐成 7位后的字符串
        tenv.executeSql("CREATE TABLE video_hbase( " +
                " video_id STRING, " +
                " f ROW<video_name STRING, video_type STRING, video_album STRING,video_author STRING, video_timelong BIGINT,create_time timestamp(3),update_time timestamp(3) >, " +
                " PRIMARY KEY (video_id) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_video_info',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");

        // 写一个insert ...select .. 的sql搞定
        tenv.executeSql(
                " INSERT INTO video_hbase        "
                        +" SELECT                         "
                        +"   lpad(cast(id as string),7,'0')  as video_id, "
                        +"   row(video_name,video_type               "
                        +"   ,video_album                 "
                        +"   ,video_author                "
                        +"   ,video_timelong              "
                        +"   ,create_time                 "
                        +"   ,update_time) as f           "
                        +" FROM video_mysql               "
        );

    }

}
